package com.bridgeintelligent.tag.bulkload.service.looklike;

import com.bridgeintelligent.tag.constants.bulkload.LookLikeMqMsg;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.mapper.IssueTaskMapper;
import com.bridgeintelligent.tag.mapper.LookLikeMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 处理looklike相关消息：根据客群ID读取es数据向hdfs服务器写文件
 */
@AllArgsConstructor
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.LOOK_LIKE_TOPIC, consumerGroup = MQConfig.LOOK_LIKE_CUSTOMERS_GROUP,consumeThreadMax=1)
public class LookLikeDelegate implements RocketMQListener<String> {

    private ObjectMapper objectMapper;

    private AsyncWriteDataService writeDataService;

    private AsyncWriteNewDataService writeNewDataService;

    private LookLikeMapper lookLikeMapper;
    
    private IssueTaskMapper issueTaskMapper;

    @SneakyThrows
    @Override
    public void onMessage(String message) {

        LookLikeMqMsg lookLikeMqMsg = objectMapper.readValue(message, LookLikeMqMsg.class);
        log.info("-------looklike生成数据文件开始：相似客群：{},数据客群：{},文件类型：{},es_index:{}",lookLikeMqMsg.getId(),lookLikeMqMsg.getCustomerId(),lookLikeMqMsg.getType(),lookLikeMqMsg.getEsIndex());
        try {
            Thread.sleep(1000);
            //幂等性控制，防止单次消费处理时间大于MQ默认15分钟触发重试机制
            int count = issueTaskMapper.transNum(lookLikeMqMsg.getUniqueIdentify());
            log.info("======>【looklike生成数据文件】当前线程：{}，处理：{}，获取count:{}",Thread.currentThread().getName(),lookLikeMqMsg.getUniqueIdentify(),count);
            if (count > 0){
                log.error("任务：{}已经处理过",lookLikeMqMsg.getUniqueIdentify());
                return;
            }
            issueTaskMapper.addTransNum(lookLikeMqMsg.getUniqueIdentify());

            writeDataService.lookalikeDateToHdfs(lookLikeMqMsg);
        } catch (Exception e) {
            log.error("消费looklike消息出现异常=======>", e);
            updateStatus(lookLikeMqMsg.getType(),AnalysisEnum.ANALYSIS_FOUR.getCode(),ChoiceEnum.CHOICE_THIRD.getCode(),lookLikeMqMsg.getId(),lookLikeMqMsg.getEsIndex());
        }
    }

    private void updateStatus(String type,String isAnalysis,String status,String id,String esIndex){
        Map<String,String> param = new HashMap<>();
        param.put("id",id);
        param.put("esIndex",esIndex);
        param.put("isAnalysis",isAnalysis);
        switch (type) {
            case "00":
                param.put("seedStatus", status);
                break;
            case "01":
                param.put("blackSeedStatus", status);
                break;
            case "02":
                param.put("whiteSeedStatus", status);
                break;
            default:
                param.put("customerSeedStatus", status);
                break;
        }
        lookLikeMapper.updateDataStatus(param);
    }


}
